package cn.aijson.demo.datasource

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.{JdbcRDD, RDD}

import java.sql.{DriverManager, ResultSet}

object Mysql {
  def main(args: Array[String]): Unit = {
    // 创建环境
    val conf: SparkConf = new SparkConf().setAppName("spark").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)
    sc.setLogLevel("WARN")

    /*建表语句
      CREATE TABLE `t_student` (
      `id` int(11) NOT NULL AUTO_INCREMENT,
      `name` varchar(255) DEFAULT NULL,
      `age` int(11) DEFAULT NULL,
      PRIMARY KEY (`id`)
      ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
       */

    //加载数据/创建RDD
    //员工集合:RDD[(部门编号, 员工姓名)]
    val empRDD: RDD[(Int, String)] = sc.parallelize(
      Seq((1011, "zhangsan"), (1012, "lisi"), (1013, "wangwu"), (1016, "lisa"))
    )

    //部门集合:RDD[(部门编号, 部门名称)]
    val deptRDD: RDD[(Int, String)] = sc.parallelize(
      Seq((1001, "销售部"), (1002, "技术部"), (1004, "客服部"))
    )

    println("------emp信息写入mysql")
    empRDD.foreachPartition(iter => {
      //开启连接
      val conn = DriverManager.getConnection("jdbc:mysql://localhost:8889/test?characterEncoding=UTF-8&useSSL=false","root","root")
      //val sql:String = "INSERT INTO `t_student` (`id`, `name`, `age`) VALUES (?, ?, null);"
      val sql:String = "INSERT INTO `t_student` (`id`, `name`) VALUES (?, ?);"
      val ps = conn.prepareStatement(sql)
      iter.foreach(t => {
        val id:Int = t._1
        val name:String = t._2
        ps.setInt(1,id)
        ps.setString(2,name)
        ps.addBatch()
        //ps.executeUpdate()
      })
      ps.executeBatch()
      //释放连接
      if (conn!=null) conn.close()
      if (ps!=null) ps.close()

    })


    println("------关联结果写入mysql")
    val empDeptRdd = empRDD.leftOuterJoin(deptRDD)

    println("------读取mysql表t_student")
    //从MySQL读取
    /*
    sc: SparkContext,
    getConnection: () => Connection, //获取连接对象的函数
    sql: String,//要执行的sql语句
    lowerBound: Long,//sql语句中的下界
    upperBound: Long,//sql语句中的上界
    numPartitions: Int,//分区数
    mapRow: (ResultSet) => T = JdbcRDD.resultSetToObjectArray _) //结果集处理函数
     */
    val getConnection =() =>  DriverManager.getConnection("jdbc:mysql://localhost:8889/test?characterEncoding=UTF-8&useSSL=false","root","root")
    val sql:String = "select id,name,age from t_student where id >= ? and id <= ?"
    val mapRow: ResultSet => (Int, String, Int) = (r:ResultSet) =>{
      val id: Int = r.getInt("id")
      val name: String = r.getString("name")
      val age: Int = r.getInt("age")
      (id,name,age)
    }

    val studentTupleRDD: JdbcRDD[(Int, String, Int)] = new JdbcRDD[(Int,String,Int)](
      sc,
      getConnection,
      sql,
      1005,
      1015,
      1,
      mapRow
    )
    studentTupleRDD.foreach(println)
    val studentRdd:RDD[(Int, String, Int)] = studentTupleRDD
  }


}
